use crate::imp::ws::websocket_imp::WsSendMessage;
use futures_channel::mpsc::UnboundedSender;
use std::sync::RwLock;
use std::time::{Duration, Instant};
use tokio::sync::watch::{Receiver, Sender};

use futures_util::{AsyncWriteExt, SinkExt, StreamExt};
// use rand::random;

lazy_static! {
    // A single-producer, multi-consumer channel that
    // only retains the last sent value.

    // when sender send a message, receiver signal.
    //channel that only retains the last sent value.
    static ref SENDER_WATCH: (
        // put a unbounded sender on watch channel
        Sender<Option<UnboundedSender<WsSendMessage>>>,
        // sender channel watch.
        Receiver<Option<UnboundedSender<WsSendMessage>>>,
    ) = tokio::sync::watch::channel(None);

    // sender message cache
    // rw lock of WsSendMessage
    static ref SENDER_MSG_CACHE: RwLock<Vec<(WsSendMessage, std::time::Instant)>> =
        RwLock::new(vec![]);
}

// set sender watch,
// put a sender_tx into SENDER_WATCH channel(mpsc channel).
pub fn set_sender_watch(
    sender_tx: Option<UnboundedSender<WsSendMessage>>,
) -> Result<(), anyhow::Error> {
    // let lock = SENDER_WATCH.read();
    if let Some(sender) = &sender_tx {
        let now = Instant::now();
        let _ = SENDER_MSG_CACHE
            .write()
            .map_err(|e| anyhow::anyhow!(e.to_string()))?
            .drain_filter(|item| {
                if now < item.1 {
                    if let Err(e) = sender.unbounded_send(item.0.clone()) {
                        log::warn!(
                            "set_sender_watch SENDER_MSG_CACHE unbounded_send err:{:?}",
                            e
                        );
                    }
                }
                true
            })
            .collect::<Vec<(WsSendMessage, Instant)>>();
    }
    SENDER_WATCH.0.send(sender_tx)?;
    Ok(())
}

// used for p2ptpnicserver.signal.rs
pub fn get_sender_receive() -> Receiver<Option<UnboundedSender<WsSendMessage>>> {
    SENDER_WATCH.1.clone()
}

// send WsSendMessage,use UnboundedSender<WsSendMessage>
pub fn send(msg: WsSendMessage) -> Result<(), anyhow::Error> {
    let receive_watch_option = SENDER_WATCH.1.borrow().clone();
    // take sender from receive_watch_option
    match receive_watch_option {
        Some(watch_sender) => {
            // get the last UnboundedSender<WsSendMessage>
            watch_sender.unbounded_send(msg)?;
        }
        None => {
            anyhow::bail!("not connect yet");
        }
    }
    Ok(())
}
// delay send wait for sender set!
pub fn send_with_cache(msg: WsSendMessage, time_out: Duration) -> Result<(), anyhow::Error> {
    let sender = SENDER_WATCH.1.borrow().clone();
    match sender {
        Some(sender) => {
            sender.unbounded_send(msg)?;
        }
        None => SENDER_MSG_CACHE
            .try_write()
            .map_err(|e| anyhow::anyhow!(e.to_string()))?
            .push((msg, Instant::now() + time_out)),
    }
    Ok(())
}
